package com.neusoft.databus.agent.interactive.collect;

import java.io.IOException;
import java.util.Optional;

import com.google.common.collect.Multimap;
import com.neusoft.bizcore.web.utils.SpringUtil;
import com.neusoft.databus.agent.parser.modbus.ModBusParser;
import com.neusoft.databus.agent.pool.netty.NettyConnector;
import com.neusoft.databus.agent.pool.netty.NettyWrapper;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.bean.MetricBean;
import com.neusoft.databus.common.connection.SocketConnection;
import com.neusoft.databus.common.parser.Parser;
import com.neusoft.databus.common.utils.DigitalUtils;

import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class NettyModbusCollector extends BasicCollector<NettyWrapper> {

    public NettyModbusCollector(Parser parser) {
        super(parser);
        this.connector = SpringUtil.getBean(NettyConnector.class);
    }

    @Override
    public void sendCommand(CommandBean command) throws InterruptedException, IOException {
        final ModBusParser mparser = (ModBusParser) this.parser;
        final Multimap<String, MetricBean> multimap = mparser.parseCollectRequest(command);
        multimap.keySet().stream().map(req -> {
            if (NettyModbusCollector.log.isDebugEnabled()) {
                NettyModbusCollector.log.debug("send a request {}", req);
            }
            return DigitalUtils.hexStringToBytes(req);
        }).forEach(adu -> {
            this.conn.getChannel().writeAndFlush(Unpooled.copiedBuffer(adu)).addListener(future -> {
                if (!future.isSuccess()) {
                    NettyModbusCollector.log.warn("failed to send {}", DigitalUtils.bytesToHexString(adu));
                }
            });
        });

        final long maxWait = System.currentTimeMillis()
                + ((((SocketConnection) command.getConnection()).getSotimeout() * multimap.keySet().size()));
        final long waitStartAt = System.currentTimeMillis();
        while ((this.conn.getPackets().size() < multimap.keySet().size()) && (System.currentTimeMillis() < maxWait)) {
            Thread.sleep(1);
        }

        if (NettyModbusCollector.log.isDebugEnabled()) {
            NettyModbusCollector.log.debug("已接收回包数量 {},耗时 {} ms", this.conn.getPackets().size(),
                    System.currentTimeMillis() - waitStartAt);
        }

        multimap.keySet().stream().forEach(req -> {
            final byte[] adu = DigitalUtils.hexStringToBytes(req);
            final Optional<byte[]> optional = this.conn.getPackets().stream()
                    .filter(it -> (adu[0] == it[0]) && (adu[1] == it[1]))
                    .findFirst();
            if (optional.isPresent()) {
                if (NettyModbusCollector.log.isDebugEnabled()) {
                    NettyModbusCollector.log.debug("{} -@- {}", req, DigitalUtils.bytesToHexString(optional.get()));
                }
                multimap.get(req).forEach(m -> m.setOriginalValue(optional.get()));
            }
        });

    }

}
